/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package brokerFailureController;

import java.util.HashMap;
import java.util.Set;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TopicSubscriber;

import momControlClient.MOMControlClient;

import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Receives messages from the detectors when they notice a change in the state
 * of a broker.
 * With that information, it deduces the state of every broker.
 * When a broker fails or recovers, it sends a message to the overlay manager
 * informing about it.
 */
public class Controller extends MOMControlClient implements MessageListener {

    /** Machine in which the controller is running. */
    private final String controllerIP;
    /** Topic used to communicate with the detector. */
    private static final String TOPIC_CONTROLLER = "brokerFailure.controller";
    /** Subscriber to listen to detectors. */
    private TopicSubscriber subscriber;
    /** Number of parts that there must be in a received message. */
    private static final int NUM_PARTS_MSG = 3;

    /** Topic used to communicate with the overlay manager. */
    private static final String TOPIC_OM = "brokerFailure.overlayManager";
    /** Publisher to send messages to the overlay manager. */
    private TopicPublisherAdapter omPublisher;

    /** List of machines in which a broker is running. */
    private static Set<String> brokers;

    /**
     * Getter for the variable 'brokers'.
     * 
     * @return variable 'brokers'
     */
    public static Set<String> getBrokers() {
        return brokers;
    }

    /**
     * Setter for the variable 'brokers'.
     * 
     * @param brokers  new value of the variable 'brokers'
     */
    public static void setBrokers(final Set<String> brokers) {
        Controller.brokers = brokers;
    }

    /**
     * List of machines in which a broker is running and their state.
     * State is an int from WORST_STATE (worst) to BEST_STATE (best).
     */
    private static HashMap<String, Integer> brokersState;

    /**
     * Getter for the variable 'brokersState'.
     * 
     * @return variable 'brokersState'
     */
    public static HashMap<String, Integer> getBrokersState() {
        return brokersState;
    }

    /**
     * Setter for the variable 'brokersState'.
     * 
     * @param brokersState  new value of the variable 'brokersState'
     */
    public static void setBrokersState(
            final HashMap<String, Integer> brokersState) {
        Controller.brokersState = brokersState;
    }

    /** Best state of a broker. */
    private static final int BEST_STATE = 10;
    /** Worst state of a broker. */
    private static final int WORST_STATE = 0;

    /**
     * List of machines in which a broker is running and the states in which
     * they detect each one of the other brokers.
     */
    private HashMap<String, HashMap<String, Integer>> brokersDetection;

    /**
     * Constructor.
     *
     * @param controllerIP     IP address in which the controller broker is
     *                         running (its own IP address)
     * @param brokerAddresses  list of brokers in the system mapped to the
     *                         ports they are listening at
     */
    public Controller(final String controllerIP,
                      final HashMap<String, Integer> brokerAddresses) {

        super(); // assign ID

        this.controllerIP = controllerIP;

        setBrokers(brokerAddresses.keySet());

        // at the beginning, assume that all brokers are off
        setBrokersState(new HashMap<String, Integer>());
        for (String broker : getBrokers()) {
            getBrokersState().put(broker, WORST_STATE);
        }

        // at the beginning, assume that all detectors detect all brokers off
        brokersDetection = new HashMap<String, HashMap<String, Integer>>();
        for (String broker : getBrokers()) {
            HashMap<String, Integer> detection
                    = new HashMap<String, Integer>();
            for (String brkr : getBrokers()) {
                detection.put(brkr, WORST_STATE);
            }
            brokersDetection.put(broker, detection);
        }

        // start a connection to read messages from the detectors and to send
        // messages to the overlay manager
        try {

            // create and start a connection and a session
            openConnAndSess(this.controllerIP);

            // create a subscriber for listening to the detectors
            subscriber = createSubscriber(TOPIC_CONTROLLER);
            // set the subscriber to listen to messages
            subscriber.setMessageListener(this);

            // create a publisher to send messages to the overlay manager
            omPublisher = createPublisher(TOPIC_OM);

            System.out.println("Controller waiting for messages in "
                               + this.controllerIP);

        } catch (Exception e) {
            System.err.println(e.getMessage());
        }

    }

    /**
     * Compares the reported state of a broker with the current one and, if it
     * is necessary, informs the overlay manager about a broker failure.
     *
     * @param host    which reported a change in a broker
     * @param broker  broker whose change is reported
     * @param state   new state of the broker who changed
     *
     * @throws JMSException  if there is any problem
     */
    public void reportChange(final String host, final String broker,
                             final int state) throws JMSException {
        System.out.println("The detector in " + host + " reported that the new"
                + " state of the broker in " + broker + " is " + state);
        // save the detected state in the detected states of the broker
        HashMap<String, Integer> states = brokersDetection.get(host);
        states.put(broker, state);
        brokersDetection.put(host, states);
        // get the current state of the broker in the controller
        int currentState = getBrokersState().get(broker);
        // if the reported state is not the current state
        if (state != currentState) {
            if (state == WORST_STATE) { // failure
                getBrokersState().put(broker, state);
                System.out.println("The broker " + broker + " has failed");
                // create text to send
                String text = "broker failed:" + broker;
                // send the message to the overlay manager
                sendMessage(omPublisher, text);
            } else if (state < currentState) { // worsening
                getBrokersState().put(broker, state);
                System.out.println("The state of the broker " + broker
                                   + " is worsening to " + state);
            } else { // improvement to the worst state detected by any detector
                     // for a broker
                Set<String> detectors = getBrokersState().keySet();
                int worstState = state;
                for (String detector : detectors) {
                    int detectedState
                            = brokersDetection.get(detector).get(broker);
                    worstState = Math.min(worstState, detectedState);
                }
                if (worstState != currentState) {
                    getBrokersState().put(broker, worstState);
                    System.out.println("The state of the broker " + broker
                                       + " is improving to " + worstState);
                }
                if (worstState == BEST_STATE) { // recovery
                    // create text to send
                    String text = "broker recovered:" + broker + ":"
                                  + worstState;
                    // send the message to the overlay manager
                    sendMessage(omPublisher, text);
                }
            }

            System.out.println("State in the controller: " + getBrokersState());
        }
        // if the reported state is the current state, do nothing
    }

    /**
     * Read the detector, broker and state reported in a received message and
     * report the change.
     *
     * This method is invoked asynchronously when a message is received.
     *
     * @param msg  received message
     */
    @Override
    public void onMessage(final Message msg) {
        try {
            JMSTextMessage message = (JMSTextMessage) msg;
            String text = message.getText();
            String[] parts = text.split(":");
            if (parts.length == NUM_PARTS_MSG) {
                reportChange(parts[0], parts[1], Integer.parseInt(parts[2]));
            } else {
                throw new Exception("Bad format of the message");
            }
        } catch (Exception e) {
            System.err.println("Error receiving or parsing message.");
        }
    }

}
